package com.example.rabbitmqspringbootdemo;

import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;
import java.util.UUID;

@Component
public class AdvancedProducer implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {
    private static final int count = 10;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /*--------------------消息可靠性发送--------------------*/
    //方法1：开启事务机制（二选一）
    //1）配置一个事务管理器Bean：RabbitTransactionManager
    //2）在发送方添加事务注解：@@Transactional
    //3）设置rabbitTemplate事务特性为true：setChannelTransacted(true)
    @Transactional     //添加事务注解
    public void transactionProducer() {
        rabbitTemplate.setChannelTransacted(true);     //设置信道为事务模式
        rabbitTemplate.convertAndSend("springTransactionQueue", "Spring事务模式---");
        // int i = 1 / 0;
    }

    //方法2：开启发送方确认机制（二选一）
    //1）在yml中设置 spring.rabbitmq.publisher-confirm-type=correlated和publisher-returns=true
    //2）定义RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback的实现类，在其中覆写confirm()和returnedMessage=()方法，处理交换器和队列异常
    //3）将rabbitTemplate中的ConfirmCallback和ReturnsCallback设置为上述实现类的实例
    public void confirmProducer() {
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnsCallback(this);
        rabbitTemplate.convertAndSend("springConfirmQueue", "Spring发送方确认模式---");
    }


    /*--------------------消息可靠性接收--------------------*/
    //1.AcknowledgeMode.NONE（autoAck或no-ack为true）：当消息被Producer发出，则自动发送ack，无需Consumer实施任何动作，也不管其是否真地收到，然后将相应消息队列中移除，存在消息丢失风险
    //2.AcknowledgeMode.AUTO(默认模式)：依据Consumer消息处理逻辑是否抛出异常自动发送ack（无异常）或nack（异常）到server端，不会丢消息
    //3.AcknowledgeMode.MANUAL（autoAck或no-ack为false）：需要人为地获取到channel之后调用方法向server发送ack（消费成功时channel.basicAck）的nack（消费失败时channel.basicReject/channel.basicNack）信息，不会丢消息
    //有ack模式(2、3)下，server端会暂存没有收到ack的消息，等消费端ack后才会丢掉；如果收到消费端的nack或connection断开没收到反馈，会将消息放回到原队列头部
    //有ack模式(2、3)下，server推送给每个channel的消息数量有限制，会保证每个channel没有收到ack的消息数量不会超过prefetchCount
    public void ackProducer() {
        for (int i = 0; i < count; i++) {
            rabbitTemplate.convertAndSend("springAckQueue", "Spring ACK模式发送数据-----" + (i + 1));
        }
    }

    //限流：在yml中设置spring.rabbitmq.listener.prefetch属性，该数值规定了可以从队列中一次性推送给消费者的最大消息数量
    //     在yml中设置spring.rabbitmq.listener.concurrency属性，该数值规定了一个消费者可以同时从队列中获取数据的并发线程数量
    public void limitProducer() {
        for (int i = 0; i < count; i++) {
            rabbitTemplate.convertAndSend("springLimitQueue", "Spring限流模式发送数据-----" + (i + 1));
        }
    }

    /*--------------------持久化--------------------*/
    public void durableProducer() {
        //Spring中Message是缺省持久化的，也可以通过构造Message来改变其相关属性
        for (int i = 0; i < count; i++) {
            Message message = MessageBuilder.withBody(("Spring 持久化发送数据-----" + (i + 1)).getBytes())
                    .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
                    .setContentEncoding("utf-8")
                    .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
                    .setMessageId(UUID.randomUUID() + "")
                    .build();
            rabbitTemplate.convertAndSend("springDurableExchange", "durableKey", message);
        }
    }

    /*--------------------死信--------------------*/
    //1）定义正常业务交换器和业务队列和routingkey，并作绑定；在队列定义时绑定死信交换器和死信deadroutingkey
    //2）定义死信交换器和死信队列和deadroutingkey，并作绑定；
    //3）定义重定向队列和redirectroutingkey，将其绑定到死信交换器
    public void deadProducer() {
        rabbitTemplate.convertAndSend(
                "springDeadExchange",
                "springDeadRoutingKey",
                10);
    }

    //--------------------延迟消息队列--------------------
    public void delayProducer() {
        System.out.println("Spring延迟消息发送时间----" + LocalDateTime.now());
        rabbitTemplate.convertAndSend(
                "springDelayExchange",
                "springDelayRoutingKey",
                "Spring 延迟消息数据-----", new MessagePostProcessor() {
                    @Override
                    public Message postProcessMessage(Message message) throws AmqpException {
                        //给消息设置延迟毫秒值
                        message.getMessageProperties().setHeader("x-delay", 5000);
                        /*等效：message.getMessageProperties().setDelay(5000);*/
                        return message;
                    }
                });

        /*等效
        MessageProperties properties = new MessageProperties();
        properties.setDelay(5000);          //properties.setHeader("x-delay", 5000);
        properties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
        Message message = MessageBuilder.withBody("Spring延迟消息数据-----".getBytes())
                .andProperties(properties)
                .build();    // Message message = new Message("Spring延迟消息数据-----".getBytes(),properties);

        rabbitTemplate.convertAndSend("springDelayExchange","springDelayRoutingKey",message);*/
    }

    //发送方确认机制回调，在publisher-confirm-type: correlated 时，任何情况都会触发本回调，不支持批量确认，
    /**
     * @param correlationData 消息信息
     * @param ack             true,MQ服务器exchange表示已经确认收到消息 ;false,没有收到消息
     * @param cause           如果没有收到消息，则为没有收到的原因，如果已经收消息到则为null
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        System.out.println("correlationData: " + correlationData);
        if (ack) {
            //成功，处理写库，更新业务
            System.out.println("消息成功到达交换器----" + correlationData.getId());
        } else {
            //失败
            System.err.println("消息未能到达交换器----" + correlationData.getId() + "-----" + cause);
        }
    }

    // 发送方确认机制回调，在publisher-returns: true 时，找到交换机了，但是没找到队列时会触发本回调
    @Override
    public void returnedMessage(ReturnedMessage returned) {
        System.err.println("消息未成功路由到队列" + returned.getMessage().getMessageProperties().getMessageId());
    }

}
